package net.bw.realtime.mall.ods.funcation;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

/**
 * @ClassName：CDCDataEtlMapFunction
 * @Author: ZddddQ
 * @Date: 2025/6/2 21:13
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class CDCDataEtlMapFunction extends RichMapFunction<String, String> {

    // 初始化
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    // 业务逻辑
    @Override
    public String map(String value) throws Exception {

        // 1.转换格式
        JSONObject jsonObject =  JSON.parseObject(value);

        // 2.获取数据
        // 获取字段值
        JSONObject beforeValue = jsonObject.getJSONObject("before");
        JSONObject afterValue = jsonObject.getJSONObject("after");
        String opValue = jsonObject.getString("op");
        Long tsValue = jsonObject.getLong("ts_ms");

        // 获取表信息
        JSONObject source = jsonObject.getJSONObject("source");
        String db_name = source.getString("db");
        String table_name = source.getString("table");

        // 判断状态
        String operatorValue;
        switch (opValue){
            case "r":
            case "u":
                operatorValue = "update";
                break;
            case "d":
                operatorValue = "delete";
                break;
            case "c":
                operatorValue = "insert";
                break;
            default:
                operatorValue = "unknown";
        }

        // 返回一个新的JSON对象
        JSONObject newJsonObject = new JSONObject();
        newJsonObject.put("db_name",  db_name);
        newJsonObject.put("table_name",  table_name);
        newJsonObject.put("operator_type",  operatorValue);

        // 判断op类型
        if("d".equals(opValue)){
            newJsonObject.put("before",  beforeValue);
        }else{
            newJsonObject.put("after",  afterValue);
        }

        newJsonObject.put("ts",  tsValue);

        // 返回字符串
        return newJsonObject.toJSONString();
    }

    // 释放资源
    @Override
    public void close() throws Exception {
        super.close();
    }
}
